package com.zhang.req.page;


import com.zhang.second.utils.UserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Timestamp;

/**
 * @title: uv统计 通过redis实现布隆过滤器
 * @author: zhang
 * @date: 2022/2/9 18:59
 */
public class UniqueVisitorCountBloomFilterWithRedis {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .readTextFile("/Users/apple/IdeaProjects/flink_1.13/src/main/resources/UserBehavior.csv")
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new UserBehavior(
                                fields[0], fields[1], fields[2], fields[3],
                                Long.parseLong(fields[4]) * 1000L
                        );
                    }
                }).filter(r -> r.type.equals("pv"))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                                    @Override
                                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                )
                .keyBy(r -> 1)
                .window(TumblingEventTimeWindows.of(Time.hours(1)))
                .trigger(new Trigger<UserBehavior, TimeWindow>() {
                    @Override
                    public TriggerResult onElement(UserBehavior element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
                        return TriggerResult.FIRE_AND_PURGE;
                    }

                    @Override
                    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                        return TriggerResult.CONTINUE;
                    }

                    @Override
                    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                        return TriggerResult.CONTINUE;
                    }

                    @Override
                    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {

                    }
                })
                .process(new ProcessWindowFunction<UserBehavior, String, Integer, TimeWindow>() {
                    private MyBloomFilter myBloomFilter;
                    private Jedis jedis;
                    private String hourUvCountKey;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        jedis = new Jedis("hadoop302",6379);
                        myBloomFilter = new MyBloomFilter(1<<30);
                        hourUvCountKey = "HourUV";
                    }

                    @Override
                    public void process(Integer integer, ProcessWindowFunction<UserBehavior, String, Integer, TimeWindow>.Context context, Iterable<UserBehavior> elements, Collector<String> out) throws Exception {

                        UserBehavior userBehavior = elements.iterator().next();
                        String time = new Timestamp(context.window().getStart()).toString();
                        String key = "BitMap_"+time;
                        long offset = myBloomFilter.getOffset(userBehavior.userId);
                        Boolean exist = jedis.getbit(key, offset);
                        if (!exist){
                            //
                            jedis.setbit(key,offset,true);
                            jedis.hincrBy(hourUvCountKey,time,1L);
                        }
                        String hget = jedis.hget(hourUvCountKey, time);
                        out.collect("窗口" + new Timestamp(context.window().getStart()) + "~" + new Timestamp(context.window().getEnd()) + "" +
                                "中的uv是：" + hget);
                    }
                })
                .print();

        env.execute();
    }


    public static class MyBloomFilter{
        private long cap;

        public MyBloomFilter(long cap) {
            this.cap = cap;
        }

        public long getOffset(String value){
            long result = 0L;
            for (char c : value.toCharArray()){
                result += result * 31 + c;
            }
            return result & (cap - 1);
        }
    }



}
